package com.unitd.scheduler.job;

import com.unitd.frame.comm.cluster.NodeNameHolder;
import com.unitd.scheduler.handler.IConfigPersistHandler;
import com.unitd.scheduler.handler.IJobLogPersistHandler;
import com.unitd.scheduler.helper.ConsistencyHash;
import com.unitd.scheduler.registry.IJobRegistry;
import com.unitd.scheduler.registry.impl.NullJobRegistry;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @desc 自定义的定时器任务的上下文对象信息
 * @filename JobContext.java
 * @copyright www.unitd.com
 * @author Hudan
 * @version 1.0
 * @date 2016/10/15
 */
public class JobContext {

	/** 时器任务的上下文对象 */
	private static JobContext context = new JobContext();
	/** 定时服务的可用节点对象 */
	private Set<String> activeNodes = new HashSet<>();
	/** 分布式状态下的一致性哈希对象 */
	private ConsistencyHash hash = new ConsistencyHash();
	/** 定时任务名称 */
	private String groupName;
	/** 所有定时任务 */
	private Map<String, AbstractJob> allJobs = new HashMap<>();
	/** 定时任务持久化处理器 */
	private IConfigPersistHandler configPersistHandler;
	/** 定时任务重试处理器 */
	private TaskRetryProcessor retryProcessor;
	/** 定时任务运行日志持久化接口 */
	private IJobLogPersistHandler jobLogPersistHandler;
	/** 定时任务执行控制 */
	private IJobRegistry registry;
	/** 定时任务同步线程 */
	private ExecutorService syncExecutor = Executors.newFixedThreadPool(1);

	public static JobContext getContext() {
		return context;
	}

	public String getGroupName() {
		return groupName;
	}

	public void setGroupName(String groupName) {
		this.groupName = groupName;
	}

	public String getNodeId() {
		return NodeNameHolder.getNodeId();
	}

	public IConfigPersistHandler getConfigPersistHandler() {
		return configPersistHandler;
	}

	public void setConfigPersistHandler(IConfigPersistHandler configPersistHandler) {
		this.configPersistHandler = configPersistHandler;
	}

	/**
	 * @desc 开启重试机制
	 */
	public void startRetryProcessor() {
		if (retryProcessor == null) {
			synchronized (context) {
				if (retryProcessor != null) return;
				retryProcessor = new TaskRetryProcessor(1);
			}
		}
	}

	public IJobLogPersistHandler getJobLogPersistHandler() {
		return jobLogPersistHandler;
	}

	public void setJobLogPersistHandler(IJobLogPersistHandler jobLogPersistHandler) {
		this.jobLogPersistHandler = jobLogPersistHandler;
	}

	public IJobRegistry getRegistry() {
		if (registry == null) {
			registry = new NullJobRegistry();
		}
		return registry;
	}

	public void setRegistry(IJobRegistry registry) {
		this.registry = registry;
	}

	public TaskRetryProcessor getRetryProcessor() {
		return retryProcessor;
	}

	public void refreshNodes(List<String> nodes) {
		activeNodes.clear();
		activeNodes.addAll(nodes);
		if (activeNodes.isEmpty()) return;
		hash.refresh(nodes);
	}

	public void addNode(String node) {
		activeNodes.add(node);
		hash.refresh(new ArrayList<>(activeNodes));
	}

	public void removeNode(String node) {
		activeNodes.remove(node);
		if (activeNodes.isEmpty()) return;
		hash.refresh(new ArrayList<>(activeNodes));
	}

	public boolean matchCurrentNode(Object shardFactor) {
		if (activeNodes.size() == 1) return true;
		String expectNodeId = hash.getAssignedRealNode(shardFactor);
		return expectNodeId.equals(getNodeId());
	}

	public void addJob(AbstractJob job) {
		String key = job.getGroup() + ":" + job.getJobName();
		allJobs.put(key, job);
	}

	public Map<String, AbstractJob> getAllJobs() {
		return allJobs;
	}

	public Set<String> getActiveNodes() {
		return activeNodes;
	}

	public void submitSyncTask(Runnable task) {
		syncExecutor.execute(task);
	}

	public void close() {
		if (retryProcessor != null) {
			retryProcessor.close();
		}
		syncExecutor.shutdown();
	}
}